package cn.ctodb.im.server.peer;

import cn.ctodb.im.api.Session;
import cn.ctodb.im.server.conf.ApplicationProperties;
import io.netty.channel.Channel;
import org.apache.kafka.clients.admin.AdminClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.*;

@Component
public class PeerManager implements InitializingBean {

    private final Logger logger = LoggerFactory.getLogger(PeerManager.class);

    private final Map<String, Peer> peerHashMap = new HashMap<>();

    private final String cacheKey = "IM_PEER";

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private ApplicationProperties applicationProperties;

    public void add(Peer peer) {
        String cid = peer.getChannel().id().asLongText();
        peerHashMap.put(cid, peer);
        getBoundHashOperations(peer.getUid()).add(peer2session(peer));
        logger.debug("用户[{}]:连接", peer.getUid());
    }

    public void remove(Channel channel) {
        try {
            Peer peer = peerHashMap.remove(channel.id().asLongText());
            if (peer != null) {
                removeByChannelIdFromCache(peer);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    public Peer get(String channelid) {
        Peer peer = peerHashMap.get(channelid);
        if (peer == null) {
//            removeByChannelIdFromCache(channelid);
        }
        return peer;
    }

    private void removeByChannelIdFromCache(Peer peer) {
        String channelid = peer.getChannel().id().asLongText();
        BoundSetOperations<String, Object> boundSetOperations = getBoundHashOperations(peer.getUid());
        Set<Object> sessions = boundSetOperations.members();
        if (sessions.size() > 0) for (Object o : sessions) {
            Session session = (Session) o;
            if (channelid.equals(session.getChannelid())) {
                boundSetOperations.remove(session);
            }
        }
        logger.debug("用户[{}:{}]:断开", peer.getUid(), channelid);
    }

    private BoundSetOperations<String, Object> getBoundHashOperations(String uid) {
        return redisTemplate.boundSetOps(cacheKey + ":" + uid);
    }

    private Session peer2session(Peer peer) {
        Session session = new Session();
        session.setChannelid(peer.getChannel().id().asLongText());
        session.setUid(peer.getUid());
        session.setGateAddr(applicationProperties.getKafkaTopic());
        return session;
    }

    @Override
    public void afterPropertiesSet() {
        // 程序退出逻辑
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            // 先处理再清空容器，否则会出bug
            peerHashMap.values().forEach(peer -> removeByChannelIdFromCache(peer));
            peerHashMap.clear();
        }, "PeerManager"));
    }
}
